package rpc.remote.transport.netty.server;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import rpc.enums.CompressTypeEnum;
import rpc.enums.RpcResponseCodeEnum;
import rpc.enums.SerializationTypeEnum;
import rpc.factory.SingletonFactory;
import rpc.remote.constants.RpcConstants;
import rpc.remote.dto.RpcMessage;
import rpc.remote.dto.RpcRequest;
import rpc.remote.dto.RpcResponse;
import rpc.remote.handler.RpcRequestHandler;

/**
 * @ClassName: NettyServiceHandler
 * @Description: 自定制ChannelHandler来处理客户端发送的数据
 * 如果继承自 SimpleChannelInboundHandler 的话就不要考虑 ByteBuf 的释放 ，{@link SimpleChannelInboundHandler} 内部的
 * channelRead 方法会替你释放 ByteBuf ，避免可能导致的内存泄露问题。详见《Netty进阶之路 跟着案例学 Netty》
 * @author: Kowk
 * @since:1.0.0
 */
@Slf4j
public class NettyRpcServerHandler extends ChannelInboundHandlerAdapter {

    private final RpcRequestHandler rpcRequestHandler;


    public NettyRpcServerHandler() {
        this.rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class);
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            if (msg instanceof RpcMessage){
                log.info("服务接受到消息[{}] ", msg);
                byte messageType = ((RpcMessage) msg).getMessageType();
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode());
                if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) {
                    rpcMessage.setMessageType(RpcConstants.HEARTBEAT_RESPONSE_TYPE);
                    rpcMessage.setData(RpcConstants.PONG);
                } else {
                    RpcRequest request = (RpcRequest) ((RpcMessage) msg).getData();
                    //执行客户端请求执行的目标方法并且返回方法结果
                    Object result = rpcRequestHandler.handle(request);
                    log.info("服务调用结果为：{} ",result.toString());
                    rpcMessage.setMessageType(RpcConstants.RESPONSE_TYPE);
                    if (ctx.channel().isActive() && ctx.channel().isWritable()){
                        RpcResponse<Object> rpcResponse = RpcResponse.success(result,request.getRequestId());
                        rpcMessage.setData(rpcResponse);
                    } else {
                        RpcResponse<Object> rpcResponse = RpcResponse.fail(RpcResponseCodeEnum.FAIL);
                        rpcMessage.setData(rpcResponse);
                        log.error("服务不活跃或不可写，调用失败消息丢弃");
                    }
                }
                ctx.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } finally {
            //确保缓冲区释放，避免造成内存泄露
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            /**
             *     ALL_IDLE : 一段时间内没有数据接收或者发送
             *     READER_IDLE ： 一段时间内没有数据接收
             *     WRITER_IDLE ： 一段时间内没有数据发送
             */
            if (state == IdleState.READER_IDLE) {
                log.info("一段数据没有接收到数据，所以关闭连接");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("服务捕捉到异常");
        cause.printStackTrace();
        ctx.close();
    }
}
